<?php

class system_mdl_queue_mysql extends base_db_model {

    function __construct($app) {
        parent::__construct($app);
        $this->db->exec('set SESSION autocommit=1;');
        $this->db->exec('set @msgID = -1;');
    }

    public function get($queue_name) {
        $time = time();
        $sql = 'UPDATE ' . DB_PREFIX . 'system_queue_mysql force index(PRIMARY) SET owner_thread_id=GREATEST(CONNECTION_ID() ,(@msgID:=id)*0),last_cosume_time=\'' . $time . '\' WHERE queue_name=\'' . $queue_name . '\' and owner_thread_id=-1 order by id LIMIT 1;';
        
        $this->db->exec($sql);
        if ($this->db->affect_row()) {
            $sql = 'select id, worker, params, create_time from ' . DB_PREFIX . 'system_queue_mysql where id=@msgID';
            $row = $this->db->selectrow($sql);
            return $row;
        }
        
        return false;
    }

    public function purge($queue_name) {
        return $this->delete(array(
            'queue_name' => $queue_name
        ));
    }

    public function is_end($queue_name) {
        if (! $this->getRow('id', array(
            'queue_name' => $queue_name,
            'owner_thread_id' => - 1
        ))) {
            return true;
        } else {
            return false;
        }
    }
}


